package streamforce.crypto.cloud.subscriber;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import streamforce.Stat;
import streamforce.crypto.KPAbe;
import streamforce.crypto.cloud.EPCloud;
import streamforce.crypto.cloud.EPCloudServer;
import streamforce.crypto.cloud.TimeTuple;
import streamforce.crypto.cloud.Tuple;
import streamforce.crypto.event.Ciphertext;
import streamforce.crypto.event.TransformedCTEventAll;
import streamforce.crypto.policies.Policy;

//for query: select price from StockEvent(cond)
//assuming that the cryptohandler has been initialized
public class CloudSumAllAgg2 extends Policy{
	final Log log = LogFactory.getLog(CloudSumAllAgg2.class);
	
	int policyId;//the same as userId	
	Stat stat,statTuple;
	public CloudSumAllAgg2(int pi, KPAbe ch){
		super(ch);
		this.policyId = pi;
		this.stat = new Stat("nTransformed");
		this.statTuple = new Stat("nTuples"); 
	}
	
	public int getPolicyId(){
		return this.policyId;
	}
	
	public void update(int timestamp, int hour, int stockId, Ciphertext ctPrice,
			Ciphertext ctVolume) {
		if (ctPrice != null && ctVolume!=null) {			
			
			byte[] ciphertextPrice = ctPrice.getCipher();
			byte[] ciphertextVolume = ctVolume.getCipher();
			long start = System.currentTimeMillis();
		
			byte[] octPrice = this.cryptoHandler.extractOriginalCipher(
					stockId, ciphertextPrice);
			byte[] tctPrice = this.cryptoHandler.transform_simple(ciphertextPrice,
					this.streamId, this.userId);						
			byte[] octVolume = this.cryptoHandler.extractOriginalCipher(
					stockId, ciphertextVolume);
			byte[] tctVolume = this.cryptoHandler.transform_simple(ciphertextVolume,
					this.streamId, this.userId);
			
			byte[][] originals = new byte[2][octPrice.length];
			byte[][] transforms = new byte[2][tctPrice.length];
			originals[0] = octPrice; transforms[0] = tctPrice;
			originals[1] = octVolume; transforms[1] = tctVolume;
		
			this.stat.update();

			long end = System.currentTimeMillis();
			log.info("agg2 transform time = " + (end - start));
			
			TimeTuple tt = EPCloudServer.tupleTime.get(new Tuple(stockId, timestamp)); 
			if (tt != null) {
				if (tt.getCount() == 0)
					this.statTuple.update();
				tt.increaseCount();
				if (EPCloudServer.QUEUE_DUMP)
					log.info(String.format("queue (%d,%d): %d", stockId,
							timestamp,
							(System.currentTimeMillis() - tt.getStart())));
			}
			
			// then construct and send another event
			TransformedCTEventAll cte = new TransformedCTEventAll(
					this.policyId, timestamp, hour, stockId, originals, transforms,
					TransformedCTEventAll.SELECT_ALL);
			EPCloud.getEPServiceProvider().getEPRuntime().sendEvent(cte);			
		}
	}
}
